/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package org.redkale.source;

import org.redkale.util.RedkaleClassLoader;

import java.net.URL;
import java.sql.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Supplier;
import java.util.logging.*;
import static org.redkale.source.DataSources.*;

/**
 *
 * <p>
 * 详情见: https://redkale.org
 *
 * @author zhangjx
 */
public class PoolJdbcSource extends PoolSource implements Supplier<Connection> {

    protected final java.sql.Driver driver;

    protected final Properties info;

    protected final String unitName;

    protected final URL persistxml;
    
    protected final ArrayBlockingQueue<Connection> queue;

    public PoolJdbcSource(String unitName, URL persistxml, Properties prop, Logger logger) {
        super(prop, logger);
        this.unitName = unitName;
        this.persistxml = persistxml;
        this.queue = new ArrayBlockingQueue<>(this.maxconns);
        this.info = new Properties();
        if (username != null) this.info.put("user", username);
        if (password != null) this.info.put("password", password);
        try {
            this.driver = DriverManager.getDriver(this.url);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    public static boolean acceptsConf(Properties property) {
        try {
            final Class driverClass = DriverManager.getDriver(property.getProperty(JDBC_URL)).getClass();
            RedkaleClassLoader.putReflectionDeclaredConstructors(driverClass, driverClass.getName());
            RedkaleClassLoader.putServiceLoader(java.sql.Driver.class);
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    @Override
    public void change(Properties property) {
        String newurl = property.getProperty(JDBC_URL, this.url);
        String newuser = property.getProperty(JDBC_USER, this.username);
        String newpassword = property.getProperty(JDBC_PWD, this.password);
        if (Objects.equals(this.url, newurl) && Objects.equals(this.username, newuser) && Objects.equals(this.password, newpassword)) return;
        try {
            this.url = newurl;
            this.username = newuser;
            this.password = newpassword;
            this.info.put("user", newuser);
            this.info.put("password", newpassword);
            logger.log(Level.INFO, DataSource.class.getSimpleName() + "(" + unitName + ") change  (" + property + ")");
        } catch (Throwable t) {
            logger.log(Level.SEVERE, DataSource.class.getSimpleName() + " dynamic change JDBC (url userName password) error", t);
        }
    }

    @Override
    public <C> void offerConnection(final C connection) {
        Connection conn = (Connection) connection;
        if (conn == null) return;
        try {
            if (checkValid(conn) && queue.offer(conn)) {
                usingCounter.decrement();
                waitingCounter.increment();
            } else {
                usingCounter.decrement();
                closeCounter.increment();
                conn.close();
            }
        } catch (Exception e) {
            logger.log(Level.WARNING, "closeSQLConnection abort", e);
        }
    }

    @Override
    public Connection get() {
        return poll();
    }

    @Override
    public CompletableFuture<Connection> pollAsync() {
        return CompletableFuture.supplyAsync(this);
    }

    @Override
    public synchronized Connection poll() {
        Connection conn = queue.poll();
        if (conn == null) {
            if (usingCounter.intValue() >= maxconns) {
                try {
                    conn = queue.poll(connectTimeoutSeconds, TimeUnit.SECONDS);
                } catch (InterruptedException t) {
                    logger.log(Level.WARNING, "take pooled connection error", t);
                }
                if (conn == null) throw new RuntimeException("create pooled connection timeout");
            }
        }
        if (conn != null) {
            usingCounter.increment();
            waitingCounter.decrement();
            if (checkValid(conn)) {
                cycleCounter.increment();
                return conn;
            } else {
                offerConnection(conn);
                conn = null;
            }
        }
        try {
            conn = driver.connect(url, info);
        } catch (SQLException ex) {
            throw new RuntimeException(ex);
        }
        usingCounter.increment();
        creatCounter.increment();
        return conn;
    }

    protected boolean checkValid(Connection conn) {
        try {
            return !conn.isClosed() && conn.isValid(1);
        } catch (SQLException ex) {
            if (!"08S01".equals(ex.getSQLState())) {//MySQL特性， 长时间连接没使用会抛出com.mysql.jdbc.exceptions.jdbc4.CommunicationsException
                logger.log(Level.FINER, "result.getConnection from pooled connection abort [" + ex.getSQLState() + "]", ex);
            }
            return false;
        }
    }

    @Override
    public void close() {
        queue.stream().forEach(x -> {
            try {
                x.close();
            } catch (Exception e) {
            }
        });
    }

}
